package com.atguigu.gmall.realtime.sink;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.util.DimUtil;
import com.atguigu.gmall.realtime.util.JdbcUtil;
import com.atguigu.gmall.realtime.util.RedisUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @Author lzc
 * @Date 2023/2/11 09:16
 */
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
    
    private Connection conn;
    private Jedis redisClient;
    
    // 建立到 phoenix 的连接
    @Override
    public void open(Configuration parameters) throws Exception {
        conn = JdbcUtil.getPhoenixConnection();
    
        redisClient = RedisUtil.getRedisClient();
    
    }
    
    // 关闭连接
    @Override
    public void close() throws Exception {
        JdbcUtil.closeConnection(conn);
    }
    
    // 流中来一条元素执行一次
    // 参数1: 流中元素
    // 参数2: 上下文
    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> t,
                       Context ctx) throws Exception {
        JSONObject data = t.f0;
        TableProcess tp = t.f1;
    
        String operateType = data.getString("operate_type");
    
        if ("delete".equals(operateType)) {
            // 删除维度信息
            deleteDataFromPhoenix(data, tp);
        }else{
            writeDataToPhoenix(data, tp);
        }
        
        // 更新 redis 缓存中的维度:
        // 1. 当维度更新或者删除的时候, 直接删除缓存中的维度
        // 2. 当维度更新,更新缓存中的维度(效率低, 需要两次连接 redis), 维度删除, 就删除缓存中的维度
        if("delete".equals(operateType) || "update".equals(operateType)){
            delDimFromRedis(data, tp);
        }
    }
    
    private void delDimFromRedis(JSONObject data, TableProcess tp) {
    
        String key = DimUtil.getRedisKey(tp.getSinkTable(), data.getString("id"));
    
        Long del = redisClient.del(key);
        if (del == 0) {
            System.out.println("你要删除的维度表: " + tp.getSinkTable() + " 的 id=" + data.getString("id") + " 的数据不存在...");
        }else{
            System.out.println("你要删除的维度表: " + tp.getSinkTable() + " 的 id=" + data.getString("id") + " 删除成功...");
        }
    }
    
    private void deleteDataFromPhoenix(JSONObject data, TableProcess tp) throws SQLException {
        // delete from user where id=?
        StringBuilder sql = new StringBuilder();
        sql
            .append("delete from ")
            .append(tp.getSinkTable())
            .append(" where id=?");
        System.out.println("phoenix 删除语句: " + sql);
        // 2. 根据连接,得到一个预处理
        PreparedStatement ps = conn.prepareStatement(sql.toString());
        // 3. 给占位符进行赋值
        // delete from user where id=?
        ps.setString(1, data.getString("id"));
        // 4. 执行 sql
        ps.execute();
        // 增删改需要 commit, 建表语句不需要
        conn.commit();
        // 6. 关闭预处理语句
        ps.close();
    }
    
    private void writeDataToPhoenix(JSONObject data, TableProcess tp) throws SQLException {
        // 1. 拼接 sql 语句
        // 主键存在就更新, 不存在就新增
        // upsert into user(a,b,c)values(?,?,?)
        StringBuilder sql = new StringBuilder();
        sql
            .append("upsert into ")
            .append(tp.getSinkTable())
            .append("(")
            .append(tp.getSinkColumns())
            .append(")values(")
            .append(tp.getSinkColumns().replaceAll("[^,]+", "?"))
            .append(")");
        
        System.out.println("phoenix 的插入语句: " + sql);
        // 2. 根据连接,得到一个预处理
        PreparedStatement ps = conn.prepareStatement(sql.toString());
        // 3. 给占位符进行赋值
        // upsert into user(a,b,c)values(?,?,?)
        String[] columns = tp.getSinkColumns().split(",");
        for (int i = 0; i < columns.length; i++) {
            String columnName = columns[i];
            String v = data.getString(columnName);
            // sql中的占位符是从 1 开始
            ps.setString(i + 1, v);
        }
        
        // 4. 执行 sql
        ps.execute();
        // 5. 提交 在 mysql 中是默认自动提交, 但是 phoenix 默认不会自动提交
        conn.commit();
        // 6. 关闭预处理语句
        ps.close();
    }
}
